// $Id: radiator.c 40 2011-04-15 15:44:23Z cedric.shih@gmail.com $
/*
 * Copyright (C) 2010 Cedric Shih <cedric.shih@gmail.com>
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>

#include "log.h"

#include "radiator.h"

#define nioradiator_lock(x) do { \
	if ((errno = pthread_mutex_lock(&x->lock)) < 0) { \
		nioerror("failed to lock: %s", strerror(errno)); \
		return NIO_ELOCK; \
	} \
} while (0)

#define nioradiator_unlock(x) do { \
	if ((errno = pthread_mutex_unlock(&x->lock)) < 0) { \
		nioerror("failed to unlock: %s", strerror(errno)); \
		return NIO_ELOCK; \
	} \
} while (0)

struct nioradiator *nioradiator_new(unsigned int concurrency)
{
	struct nioradiator *radiator;

	niodebug("constructing");

	if (!(radiator = calloc(1, sizeof(struct nioradiator)))) {
		nioerror("failed to callloc radiator");
		goto error;
	}

	TAILQ_INIT(&radiator->futures);
	radiator->level = concurrency;

	return radiator;

error:
	nioradiator_free(radiator);
	return NULL;
}

void nioradiator_free(struct nioradiator *radiator)
{
	if (!radiator) {
		return;
	}

	niodebug("destructing: %p", radiator);

	free(radiator);
}

int nioradiator_start(struct nioradiator *radiator)
{
	int rc;

	if (!radiator) {
		nioerror("null pointer");
		rc = NIO_EINVAL;
		goto finally;
	}

	if (radiator->started) {
		nioerror("already started");
		rc = NIO_EPERM;
		goto finally;
	}

	niodebug("starting");

	if ((rc = pthread_mutex_init(&radiator->lock, NULL))) {
		nioerror("failed to init mutex: %s", strerror(rc));
		goto finally;
	}

	if ((rc = pthread_cond_init(&radiator->start, NULL))) {
		nioerror("failed to init cond: %s", strerror(rc));
		goto finally;
	}

	if ((rc = pthread_cond_init(&radiator->job, NULL))) {
		nioerror("failed to init cond: %s", strerror(rc));
		goto finally;
	}

	if (!(radiator->threads = calloc(radiator->level, sizeof(pthread_t)))) {
		nioerror("failed to calloc %d of pthread_t", radiator->level);
		rc = NIO_ENOMEM;
		goto finally;
	}

	if ((rc = nioradiator_start_threads(radiator))) {
		nioerror("failed to start threads");
		goto finally;
	}

	radiator->started = 1;
	rc = 0;

finally:
	return rc;
}

int nioradiator_stop(struct nioradiator *radiator)
{
	int rc;

	if (!radiator) {
		nioerror("null pointer");
		rc = NIO_EINVAL;
		goto finally;
	}

	if (!radiator->started) {
		nioerror("not started");
		rc = NIO_EPERM;
		goto finally;
	}

	niodebug("stopping");

	radiator->started = 0;

	if ((rc = nioradiator_stop_threads(radiator))) {
		nioerror("failed to stop threads");
		goto finally;
	}

	free(radiator->threads);

	if ((rc = pthread_cond_destroy(&radiator->job))) {
		nioerror("failed to destroy cond: %s", strerror(rc));
		goto finally;
	}

	if ((rc = pthread_cond_destroy(&radiator->start))) {
		nioerror("failed to destroy cond: %s", strerror(rc));
		goto finally;
	}

	if ((rc = pthread_mutex_destroy(&radiator->lock))) {
		nioerror("failed to destroy mutex: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	return rc;
}

int nioradiator_start_threads(struct nioradiator *radiator)
{
	int rc;
	unsigned int i;

	niodebug("starting threads");

	nioradiator_lock(radiator);

	for (i = 0; i < radiator->level; i ++) {
		niodebug("creating thread %d", i);

		if ((rc = pthread_create(radiator->threads + i, NULL,
				nioradiator_cb, radiator))) {
			nioerror("failed to create thread %d: %s",
					i, strerror(rc));
			goto finally;
		}

		niodebug("detaching thread %d", i);

		if ((rc = pthread_detach(radiator->threads[i]))) {
			nioerror("failed to detach thread %d: %s",
					i, strerror(rc));
			goto finally;
		}
	}

	niodebug("waiting for all threads started");

	if ((rc = pthread_cond_wait(&radiator->start, &radiator->lock))) {
		nioerror("failed to wait: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	nioradiator_unlock(radiator);
	return rc;
}

int nioradiator_stop_threads(struct nioradiator *radiator)
{
	int rc;

	niodebug("stopping threads");

	nioradiator_lock(radiator);

	radiator->started = 0;

	niodebug("broadcasting stopping");

	if ((rc = pthread_cond_broadcast(&radiator->job))) {
		nioerror("failed to broadcast: %s", strerror(rc));
		goto finally;
	}

	niodebug("waiting for all threads stopped");

	if ((rc = pthread_cond_wait(&radiator->start, &radiator->lock))) {
		nioerror("failed to wait: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	nioradiator_unlock(radiator);
	return rc;
}

int nioradiator_dispatch(struct nioradiator *radiator, struct niofuture *future)
{
	int rc;

	niodebug("dispatching: %p", future);

	nioradiator_lock(radiator);

	niodebug("enqueue event");

	TAILQ_INSERT_TAIL(&radiator->futures, future, next);

	niodebug("signaling to process event");

	if ((rc = pthread_cond_signal(&radiator->job))) {
		nioerror("failed to signal: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	nioradiator_unlock(radiator);
	return rc;
}

void *nioradiator_cb(void *arg)
{
	nioradiator_do(arg);
	return NULL;
}

int nioradiator_do(struct nioradiator *radiator)
{
	int rc;
	struct niofuture *future;

	nioradiator_lock(radiator);

	radiator->active++;

	if (radiator->active >= radiator->level) {
		niodebug("signaling for start");

		if ((rc = pthread_cond_signal(&radiator->start))) {
			nioerror("failed to signal: %s", strerror(rc));
			goto finally;
		}
	}

	while (1) {
		if (TAILQ_EMPTY(&radiator->futures)) {
			niodebug("waiting for event");

			if ((rc = pthread_cond_wait(&radiator->job,
					&radiator->lock))) {
				nioerror("failed to wait: %s", strerror(rc));
				break;
			}
		}

		if (!radiator->started) {
			niodebug("breaking thread loop");
			break;
		}

		niodebug("dequeueing event");

		future = TAILQ_FIRST(&radiator->futures);
		TAILQ_REMOVE(&radiator->futures, future, next);

		if ((rc = pthread_mutex_unlock(&radiator->lock)) < 0) {
			nioerror("failed to unlock: %s", strerror(rc));
			goto finally;
		}

		niodebug("invoking: %p", future->cb);
		(*future->cb)(future, future->socket,
				future->flags, future->cbarg);

		if ((rc = pthread_mutex_lock(&radiator->lock)) < 0) {
			nioerror("failed to lock: %s", strerror(rc));
			goto finally;
		}
	}

	rc = 0;

finally:
	radiator->active--;

	if (radiator->active <= 0) {
		niodebug("signaling for stop");

		if ((rc = pthread_cond_signal(&radiator->start))) {
			nioerror("failed to signal cond: %s", strerror(rc));
		}
	}
	nioradiator_unlock(radiator);
	return rc;
}
